package hyl.base.mq.kafka;

import java.util.Properties;

import hyl.core.MyFun;
import hyl.core.conf.MyPrp;

public class KfkConfig {

	private static MyPrp _prpp;
	private static MyPrp _prps;
	private static String _brokerAddr = "localhost:9092";
	// private static String _zkAddr = "localhost:9092";
	private static String _groupid = "test";
	public static String getSub(String key) {
		return _prps.get(key);
	}
	public static String getPub(String key) {
		return _prpp.get(key);
	}
	public static Properties getCfgPub() {
		if (_prpp == null) {
			_prpp = new MyPrp("mq/kafka_pub.conf");
			_prpp.set("bootstrap.servers", _brokerAddr);
			/**
			 * ack是判别请求是否为完整的条件（就是是判断是不是成功发送了）。 我们指定了“all”将会阻塞消息，这种设置性能最低，但是是最可靠的。
			 */
			_prpp.set("acks", "all");
			/**
			 * retries，如果请求失败，生产者会自动重试， 我们指定是0次，如果启用重试，则会有重复消息的可能性。
			 */
			_prpp.set("retries", "0");
			/**
			 * producer(生产者)缓存每个分区未发送消息。缓存的大小是通过 batch.size 配置指定的。
			 * 值较大的话将会产生更大的批。并需要更多的内存（因为每个“活跃”的分区都有1个缓冲区）。
			 */
			_prpp.set("batch.size", "4096");
			/**
			 * 即便分区的缓冲空间还没有满，默认缓冲也可立即发送.但是 ，如果你想减少请求的数量，可以设置linger.ms大于0。
			 * 这将指示生产者发送请求之前等待一段时间，希望更多的消息填补到未满的批中。 这类似于TCP的算法，例如上面的代码段，可能100条消息在一个请求发送，
			 * 因为我们设置了linger(逗留)时间为2毫秒，然后，如果我们没有填满缓冲区， 这个设置将增加1毫秒的延迟请求以等待更多的消息。需要注意的是，在高负载下，
			 * 即使是 linger.ms=0,也会组成批。在不处于高负载的情况下， 如果设置比0大，以少量的延迟代价换取更少的，更有效的请求。
			 * 
			 */
			_prpp.set("linger.ms", "1");
			/**
			 * 同步或异步
			 * 
			 */
			_prpp.set("producer.type ", "async");
	
			/**
			 * 控制生产者可用的缓存总量 33554432，如果消息发送速度比其传输到服务器的快， 将会耗尽这个缓存空间。当缓存空间耗尽，其他发送调用将被阻塞，
			 * 阻塞时间的阈值通过max.block.ms设定，之后它将抛出一个TimeoutException。
			 */
			_prpp.set("buffer.memory", "65535");
			/**
			 * 将用户提供的key和value对象ProducerRecord转换成字节，你可以使用附带
			 * 的ByteArraySerializaer或StringSerializer处理简单的string或byte类型。
			 */

			_prpp.set("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
			_prpp.set("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
		}
		return _prpp.getAll();
	}

	public static Properties getCfgSub(String 客户端编号) {
		if (_prps == null) {
			_prps = new MyPrp("mq/kafka_sub.conf");
			_prps.set("bootstrap.servers", "localhost:9092");
			// 消费者所在组
			_prps.set("group.id", _groupid);

			_prps.set("client.id", 客户端编号);
			// zk连接超时
//		_prps.set("zookeeper.connect", _zkAddr);
//		_prps.set("zookeeper.session.timeout.ms", "4000");
//		_prps.set("zookeeper.sync.time.ms", "200");	
			// _prps.set("auto.offset.set", "smallest");
			// 是否自动提交
			_prps.set("enable.auto.commit", "false");

			isautocommit = MyFun.str2Bool(_prps.get("enable.auto.commit"));
			// _prps.set("auto.commit.interval.ms", "1000");
			_prps.set("session.timeout.ms", "30000");
			// 序列化类
			_prps.set("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
			_prps.set("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
			// _prps.set("serializer.class", "kafka.serializer.StringEncoder");

		}

		return _prps.getAll();
	}

	/** 是否自动提交 */
	private static boolean isautocommit = true;

	/** 是否自动提交 */
	public static boolean isAutoCommit() {
		return isautocommit;
	}

}
